package com.zzw.demo.util;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

public class KafkaUtil {


    public static FlinkKafkaProducer<String> kafkaProducer( String topic){
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        FlinkKafkaProducer<String> myProducer = new FlinkKafkaProducer<>(
                topic,
                new SimpleStringSchema(),
                properties);

        return myProducer;
    }

    public  static FlinkKafkaConsumer<String> kafkaSource(String topic,String group_id) throws Exception {

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop-single:9092");
        properties.setProperty("group.id",group_id);
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), properties);

        return consumer;
    }











}
